package org.example.data;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.example.cdc.Main;

public class Convert {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        //创建一个DataStream
        DataStreamSource<String> dataStream = env.fromElements("hello", "world", "json");
        Table inputTable = tEnv.fromDataStream(dataStream);
        //注册一个视图以供查询
        tEnv.createTemporaryView("inputTable", inputTable);
        Table resultTable = tEnv.sqlQuery("SELECT UPPER(f0) from inputTable");

        DataStream<Row> resultStream = tEnv.toDataStream(resultTable);
        resultStream.print();
        env.execute("Convert");


    }
}
